#include "util.h"
using namespace std;
typedef struct _arg_t_
{
	int no;
}arg_t;
//std::ofstream stresslog("./stresslog.txt");

void *getRecord(void* arg)
{
    char buffer[SIZE];
    for(unsigned int i=0;Request::stressIsRunning;i++)
    {
        if(i==Request::vecData.size())
            i=0;
        //cout<<"i\t"<<i<<endl;
        if(feof(Request::fin))
            rewind(Request::fin);
        //cout<<"isFull"<<Request::vecData[i].isFull<<"\ti\t"<<i<<endl;
        if(Request::vecData[i].isFull==false)
        {
            //cout<<"index\t"<<i<<"\tsize\t"<<Request::vecData.size()<<endl;
            Request::vecData[i].data.clear();
            while(fgets(buffer,SIZE,Request::fin))
            {
                if(buffer[strlen(buffer)]=='\0'&&buffer[strlen(buffer)-1]=='\n')
                {
                    buffer[strlen(buffer)-1]='\0';
                    Request::vecData[i].data=Request::vecData[i].data+buffer;
                    Request::vecData[i].isFull=true;
                    //cout<<"change "<<i<<" to ture"<<endl;
                    break;
                }
                else
                {
                    Request::vecData[i].data=Request::vecData[i].data+buffer;
                }
            }
        }
    }
}


void *control(void* arg)
{
    sleep(Request::stressInterval);
    Request::stressIsRunning=false;
    return (void *)NULL;
}

char *mytime(struct timeval &tv)
{
    char outstr[200];
    time_t t;
    struct tm *tmp;
    //gettimeofday(&tv,NULL);
    tmp = localtime(&(tv.tv_sec));
    if (tmp == NULL) { 
        perror("localtime");
        exit(EXIT_FAILURE);
    } 
    //cout<<"localtime\t"<<tmp<<endl;   
    if (strftime(outstr, sizeof(outstr),"%D %T", tmp) == 0)  
    {   
        fprintf(stderr, "strftime returned 0");
        exit(EXIT_FAILURE);
    }   
    return strdup(outstr);
}

int work(int argc,char *argv[])
{
    //FILE *fin=NULL;
    char buffer[SIZE];
    //string data;
    //cout << "Usage:"<< argv[0]<<"threadnum interval page logfile ip port queryFile" <<endl;
    //./client 1 10 /esmart ./log.txt 10.20.146.85 8080 ~/prophet_head.txt 
    Request::stressThreadnum=atol(argv[1]);
    Request::stressInterval=atol(argv[2]);
    Request::strPage=argv[3];
    Request::strLogName=argv[4];
    Request::strIp=argv[5];
    //cout<<"ip\t"<<Request::strIp<<endl;
    Request::intPort=atol(argv[6]);
    Request::strFileName=argv[7];
    Request::stresslog.open(Request::strLogName.c_str());
    //if((fin=fopen(Request::strFileName.c_str(),"r"))==NULL)
    Request::openFile();
    //Request::vecData.push_back(0);
    long l=0,prePos=0,nextPos=0;
    struct timeval tv1,tv2;
    Request::ratio=32;
    //Request::vecData.clear();
    //Request::vecData.resize(Request::stressThreadnum*Request::ratio);
    unsigned int lineno=0;
    //string _data;
    data_t data;
    while(fgets(buffer,SIZE,Request::fin)&&(lineno<(Request::stressThreadnum*Request::ratio)))
    {
        if(buffer[strlen(buffer)]=='\0'&&buffer[strlen(buffer)-1]=='\n')
        {
            buffer[strlen(buffer)-1]='\0';
            data.data=data.data+buffer;
            data.isFull=true;
            Request::vecData.push_back(data);
            //cout<<"lineno\t"<<lineno<<"\tsize of Request::vecData[lineno]\t"<<Request::vecData[lineno].data.size()<<"\tstressThreadnum*ratio\t"<<Request::stressThreadnum*Request::ratio<<endl;
            data.data.clear();
            lineno++;
            Request::stressCount++;
        //    if(Request::stressCount==Request::stressThreadnum*Request::ratio) break;
        }
        else
        {
            //l=l+SIZE;
            //l=l+strlen(buffer);
            //Request::vecData[lineno].data=Request::vecData[lineno].data+buffer;
            data.data=data.data+buffer;
        }
        //memset(buffer,0,SIZE);
    }
    //Request::vecData.pop_back();

    Request::stresslog<<"get total lines:"<<Request::stressCount<<"\tvecData size:\t"<<Request::vecData.size()<<endl;
    cout<<"get total lines:"<<Request::stressCount<<"\tvecData size:\t"<<Request::vecData.size()<<endl;
    //cout<<"get total lines:"<<Request::stressCount<<endl;
    //printf("get total lines:%lu",Request::stressCount);
    if(Request::stressCount<Request::stressThreadnum)
    {
        cout<<"Total lines less than Request::stressThreadnum,exit"<<endl;
        return 1;
    }
    int i=0;
    pthread_t pid[Request::stressThreadnum],conid,recordid;
    pthread_create(&conid,NULL,control,NULL);
    pthread_create(&recordid,NULL,getRecord,NULL);
    arg_t tharg[Request::stressThreadnum];
    //struct timeval start,end;
    gettimeofday(&Request::start,NULL);
    //cout<<ctime(&(Request::start.tv_sec))<<endl;
    for(int j=0;j<Request::stressThreadnum;j++)
    {
        Request::stresslog<<"Thread created,   number:"<<j<<endl;
        tharg[j].no=j;
        pthread_create(&pid[j],NULL,stress,(void*)&(tharg[j]));
    }
    Request::stresslog<<"All threads have created,waiting for running"<<endl;
    cout<<"All threads have created,waiting for running"<<endl;
    unsigned long long total=0;
    unsigned long long cortotal=0;
    unsigned long long errtotal=0;
    unsigned long long errConTotal=0;
    pid_t proid=getpid();
    char cmd[1024];
    snprintf(cmd,1024,"cat /proc/%d/status |egrep \"VmSize|VmRSS|VmData\"",proid);
    system(cmd);
    pthread_join(recordid,NULL);
    for(int j=0;j<Request::stressThreadnum;j++)
    {
        Request::stresslog<<"Thread destroyed, number :"<<j<<endl;
        void *void_ptr;
        pthread_join(pid[j],&void_ptr);
        ret_t value_ptr=*((ret_t *)void_ptr);
        Request::stresslog<<"value_ptr total    : "<<(uint64_t)(value_ptr.total)<<" in thread :"<<j<<endl;
        Request::stresslog<<"value_ptr cottotal : "<<(uint64_t)(value_ptr.cortotal)<<" in thread :"<<j<<endl;
        Request::stresslog<<"value_ptr errtotal :"<<(uint64_t)(value_ptr.errtotal)<<" in thread :"<<j<<endl;
        //cout<<"value_ptr total    : "<<value_ptr.total<<" in thread :"<<j<<endl;
        //cout<<"value_ptr cottotal : "<<value_ptr.cortotal<<" in thread :"<<j<<endl;
        //cout<<"value_ptr errtotal :"<<value_ptr.errtotal<<" in thread :"<<j<<endl;
        Request::ret.total=Request::ret.total+value_ptr.total;
        Request::ret.errtotal=Request::ret.errtotal+value_ptr.errtotal;
        Request::ret.errConTotal=Request::ret.errConTotal+value_ptr.errConTotal;
        Request::ret.cortotal=Request::ret.cortotal+value_ptr.cortotal;
        delete (ret_t *)void_ptr;
    }
    pthread_join(conid,NULL);
    gettimeofday(&Request::end,NULL);
    cout<<ctime(&(Request::end.tv_sec))<<endl;
    showResult();
}
void showResult()
{
    Request::stresslog<<"Total queries executed: "<<Request::ret.total<<" process time,seconds: "<<Request::end.tv_sec-Request::start.tv_sec<<"  microseconds: "<<Request::end.tv_usec-Request::start.tv_usec<<endl;
    Request::stresslog<<"ERRTotal queries executed: "<<Request::ret.errtotal<<" process time,seconds: "<<Request::end.tv_sec-Request::start.tv_sec<<"  microseconds: "<<Request::end.tv_usec-Request::start.tv_usec<<endl;
    Request::stresslog<<"CORTotal queries executed: "<<Request::ret.cortotal<<" process time,seconds: "<<Request::end.tv_sec-Request::start.tv_sec<<"  microseconds: "<<Request::end.tv_usec-Request::start.tv_usec<<endl;
    Request::stresslog<< "Average Time-consuming microseconds Per Query: " << (Request::end.tv_sec - Request::start.tv_sec)*1000000/(double)Request::ret.total << endl;
    Request::stresslog<< "Average Queries Per Second: " << (double)(Request::ret.total)/(Request::end.tv_sec - Request::start.tv_sec) << endl;
    cout<<endl;
    cout<<" --- FROM("<<mytime(Request::start)<<") TO("<<mytime(Request::end)<<") TIME-USED("<<Request::end.tv_sec-Request::start.tv_sec<<" seconds) ---"<<endl;
    cout<<"QPS:                      "<<(double)Request::ret.total/(Request::end.tv_sec - Request::start.tv_sec)<<endl;
    cout<<"Latency:                  "<< (Request::end.tv_sec - Request::start.tv_sec)*1000000/(double)Request::ret.total<<"ms"<<endl;
    cout<<"Lines:                    "<<Request::stressCount<<endl;
    cout<<"Query Success Number:     "<<Request::ret.cortotal<<endl;
    cout<<"Query Failed Number:      "<<Request::ret.errtotal<<endl;
    cout<<"Connect Failed Number:    "<<Request::ret.errConTotal<<endl;
    printf("Min Response Time:        %.1f ms\n", (float)Request::_intMinResponseTime/1.0);
    printf("Max Response Time:        %.1f ms\n", (float)Request::_intMaxResponseTime/1.0);
    int totNum = 0;
    int percent[] = {25, 50, 75, 90, 95, 99};
    int index = 0;
    float maxTime = 0.0;
    int lastNum = 0;
    map<int,int>::const_iterator it;
    for(it=Request::_responseTimeMap.begin();index < 6 && it!=Request::_responseTimeMap.end(); it++) {
        totNum += it->second;
        //cout<<"totNum:\t"<<totNum<<"\tindex"<<index<<"\t"<<percent[index]<<"\tcortotal"<<cortotal<<"\tpercentage:\t"<<(1.0*cortotal)/100*percent[index] <<endl;
        if (totNum > 0 && totNum >= 
                    (1.0*Request::ret.cortotal)/100*percent[index]) 
        {
            if(lastNum != totNum) {
                maxTime = (float)(it->first)/1.0;
                lastNum = totNum;
            }
            printf("%d Percentile:            %.1f ms\n", percent[index], maxTime);
            index ++;
            if (index > 5) break;
        }
    }
    for (; index <= 5; index++) {
        printf("%d Percentile:            %.1f ms\n", percent[index], maxTime);
    }
}
/*
* --- FROM(07/17/12 11:38:48) TO(07/17/12 12:38:48) TIME-USED(1h-0min-0s) ---
CMD: abench_2.1.1 -p 8 -s 3600 --http -o log.txt -k 10.20.146.94 38811 /home/admin/newQuery
QPS:                           311.72
Latency:                       25.60 ms
Query Success Number:          1122181
Connect Failed Number:         0
Query Failed Number:           6
Query Timeout Number:          0
Match Failed Number:           0
No DocsReturn Number:          406057 (36.18%)
Average DocsReturn Number:     28.45
No DocsFound Number:           401889 (35.81%)
Average DocsFound Number:      1832.06
Average Return Length(Bytes):  10206
Min Response Time:             0.4 ms
Max Response Time:             1047.5 ms
25 Percentile:                 7.1 ms
50 Percentile:                 14.7 ms
75 Percentile:                 32.5 ms
90 Percentile:                 55.6 ms
95 Percentile:                 76.3 ms
99 Percentile:                 169.4 ms
*/


void *stress(void* arg)
{
    arg_t *ptr=(arg_t*)arg;
    Request::stresslog<<"Thread no:        "<<ptr->no<<endl;
    Request::stresslog<<"Mycount:          "<<Request::stressCount<<endl;
    Request::stresslog<<"Total thread num: "<<Request::stressThreadnum<<endl;
    int index=ptr->no;
    uint32_t ret;
    Request::stresslog<<"index:"<<index<<endl;
    //cout<<"Thread-Index:\t"<<index<<endl;
    unsigned long long total=0,cortotal=0,errtotal=0,errConTotal;
    //Request request(Request::strFileName);
    Request request;
    for(int i=index;Request::stressIsRunning;i=i+Request::stressThreadnum)
    {
        if(Request::vecData[i].isFull==false) continue;
        if(i>=Request::stressCount)
        {
            i=index;
        }
        //cout<<"processing\t"<<i<<endl;
        ret=request.work(i);
        if(ret==0) 
        {
            cortotal++;
            Request::vecData[i].isFull=false;
        }
        else if(ret==-1)
        {
             errtotal++;
             Request::stresslog<<"thread["<<index<<"]:\t"<<Request::vecData[i].data<<endl;
        }
        else if(ret==-2) 
        {
            errConTotal++;
        }
        total++;
    }
    ret_t tret,*pret=new ret_t();
    tret.total=total;
    tret.cortotal=cortotal;
    tret.errtotal=errtotal;
    tret.errConTotal=errConTotal;
    Request::stresslog<<"total query "<<(uint64_t)tret.total<<" in thread no: "<<index<<endl;
    Request::stresslog<<"total correct query "<<(uint64_t)tret.cortotal<<" in thread no: "<<index<<endl;
    Request::stresslog<<"total error query "<<(uint64_t)tret.errtotal<<" in thread no: "<<index<<endl;
    //cout<<"total query "<<(uint64_t)tret.total<<" in thread no: "<<index<<endl;
    //cout<<"total correct query "<<(uint64_t)tret.cortotal<<" in thread no: "<<index<<endl;
    //cout<<"total error query "<<(uint64_t)tret.errtotal<<" in thread no: "<<index<<endl;
    pret->total=total;
    pret->errtotal=errtotal;
    pret->errConTotal=errConTotal;
    pret->cortotal=cortotal;
    return ((void*)pret);
}


void SignHandler(int iSignNo)
{
    //printf("Capture sign no:%d/n",iSignNo); 
    
    gettimeofday(&Request::end,NULL);
    cout<<ctime(&(Request::end.tv_sec))<<endl;
    showResult();
    exit(1);
}


int main(int argc, char* argv[])
{
    cout<<"CMD:\t";
    for(unsigned int i=0;i<argc;i++)
        cout<<argv[i]<<" ";
    signal(SIGINT,SignHandler);
    struct timeval now;
    gettimeofday(&now,NULL);
    cout<<endl<<"START TIME:\t"<<mytime(now)<<endl;
    if (8 != argc)
    {
        //./abench -p 10 -s 10  --http /esmart -o log.txt -k 10.0.0.0 12345 ~/queryDir
        //./client 1 10 /esmart ./log.txt 10.20.146.85 8080 ~/prophet_head.txt 
        cout << "Usage:"<< argv[0]<<"threadnum interval page logfile ip port queryFile" <<endl;
        return -1;
    }
    cout << argv[1] <<" MultiHttpPress test start...\n" <<  endl;
    work(argc,argv);
    cout << argv[1] <<" MultiHttpPress test end!\n" <<  endl;
    return 0;
}






    /*
    while(fgets(buffer,SIZE,fin))
    {
        if(buffer[strlen(buffer)]=='\0'&&buffer[strlen(buffer)-1]=='\n')
        {
            buffer[strlen(buffer)-1]='\0';
            data=data+buffer;
            Request::vecData.push_back(data);
            cout << "vecdata::size()\t"<<Request::vecData.size()<<endl;
            data.clear();
        }
        else
        {
            data=data+buffer;
        }
    }
    cout << "vecdata::size()\t"<<Request::vecData.size()<<endl;
    */
